package main

import (
	"errors"
	"fmt"
	"math/rand"
	"os"
	"pkg/yh_config"
	"pkg/fmtlog"
	"pkg/libnet"
	"runtime"
	"runtime/debug"
	"strconv"
	"strings"
	"sync"
	"sync/atomic"
	"time"
	"sample/adapter/config"
	"sample/pkg/libnet"
	"sample/adapter/service"
)

const (
	CLIENT_MAXAGENT_SIZE      = 5000    //最大客户端连接数
	CLIENT_CHANNEL_SIZE       = 10      //客户端Channel的容量
	SERVER_CHANNEL_SIZE       = 10000   //服务端Channel的容量
	MAX_MSG_CONTENT_LEN       = 2097152 //消息体的最大长度
	LOG_INTERVAL              = 10      //状态日志输出的间隔
	CLIENT_FIRST_READ_TIMEOUT = 10      //客户端连接后第一条消息发送到服务器的超时时间
	SERVER_READ_TIMEOUT       = 300     //服务器idle的最长时间
	//SERVER_MSG_CONSUME_INTERVAL = 5       //服务器检查Channel中消息是否过多的时间间隔
	WAIT_LAST_CONN_TIMEOUT = 60 //等待同一用户上一个连接彻底关闭的超时时间
)

var rts, rt, wt, rlock, wlock int32
var waiting int32

//日志
var glog *fmtlog.FmtLog

//配置
var conf *yh_config.Config

//心跳消息的间隔
var clientReadTimeout uint //客户端idle的最长时间

//丢弃的消息数量
var dropMsgs uint = 0
var dropServerMsgs uint = 0

//用户上行消息通道
var servicesMap map[uint16][]*service.ServiceConn

//服务器下行消息通道表
var clientsMap map[uint32](*service.ServiceConn) = make(map[uint32](*service.ServiceConn))

var clientMutex sync.RWMutex

//是否需要向Server发送断开连接消息
var needSendCloseMsg map[uint32]bool = make(map[uint32]bool)
var needSendCloseLock sync.Mutex

//配置文件中的必要项
var keywords = map[string]bool{
	"ips":       true,
	"port":      true,
	"log":       true,
	"log_level": true,
	"procs":     true,
	"servers":   true,
	"timeout":   true,
}

var conn_chan = make(chan *yh_net.TCPConn)
var ch_conn_change = make(chan int)
var cur_conn_num = 0

//检查配置文件是否合法
func checkConfig(conf *yh_config.Config) error {
	for key := range keywords {
		if _, found := conf.Items[key]; found == false {
			return errors.New("not found key [" + key + "]")
		}
	}
	return nil
}

//输出系统状态
func logStatus() {
	for {
		time.Sleep(LOG_INTERVAL * time.Second)
		log := ""
		for t, servers := range servicesMap {
			for id, ch := range servers {
				log += fmt.Sprintf("%v-%v:%v\t", t, id, len(ch.Ch))
			}
		}
		glog.Debugf("Messages in server channel : "+log)
		glog.Debugf("online clients : %v\n\tread from client routings : %v\n\twrite to client routings : %v\n\twrite to server routings : %v | waiting : %v rlock: %v wlock: %v", len(clientsMap), rt, wt, rts, waiting, rlock, wlock)
		glog.Info("drop messages : from clients %v, from servers %v", dropMsgs, dropServerMsgs)
		dropMsgs, dropServerMsgs = 0, 0
	}
}

func closeClient(id uint32) {
	clientMutex.Lock()
	wlock++
	defer func() { wlock--; clientMutex.Unlock() }()
	c, ok := clientsMap[id]
	if ok {
		c.Close()
		delete(clientsMap, id)
	}
	needSendCloseLock.Lock()
	v, ok := needSendCloseMsg[id]
	needSendCloseLock.Unlock()
	if !ok || v {
		e := service.NewMsg(0)
		e.SetID(id)
		e.SetLength(0)
		servicesMap[0][id%uint32(len(servicesMap[0]))].Ch <- e
	}
	needSendCloseLock.Lock()
	delete(needSendCloseMsg, id)
	needSendCloseLock.Unlock()
}

func ReadMsgs(conn *libnet.TCPConn) {
	atomic.AddInt32(&rt, 1)
	defer atomic.AddInt32(&rt, -1)
	defer conn.Close()
	//第一条消息要在10秒内收到，否则直接断开，避免攻击
	conn.SetReadTimeout(CLIENT_FIRST_READ_TIMEOUT)
	head := service.NewMsg(0)
	first := true
	second := true
	var newKey uint32 = 0
	var uid uint32

	for {
		err := conn.ReadSafe(head.Header())
		if err != nil {
			fmt.Printf("user %v ReadSafe head error: %v\n", uid, err.Error())
			break
		}
		if head.Length() > MAX_MSG_CONTENT_LEN {
			fmt.Printf("from %v len=%d > max id=%d type=%d", uid, head.Length(), head.ID(), head.Type())
			break
		}

		data := service.NewMsg(head.Length())
		err = conn.ReadSafe(data.Content())
		if err != nil {
			fmt.Printf("user %v ReadSafe content error: %v", uid, err.Error())
			break
		}

		copy(data.Header(), head.Header())
		if data.Type() == service.HEARTBEAT_MSG {
			fmt.Printf("receive from client len=%d id=%d type=HEART-BEAT", head.Length(), head.ID())
		} else {
			fmt.Printf("receive from client len=%d id=%d type=%d content=%s", head.Length(), head.ID(), head.Type(), string(data.Content()))
		}

		if second {
			if first {
				if data.Type() != service.HEARTBEAT_MSG {
					fmt.Printf("user %v connected, but first message is not heart beat, abort.", head.ID())
					break
				}
				oldKey, newKeyTmp := data.HeartBeatContent()
				newKey = newKeyTmp
				fmt.Printf("user %v oldKey=%v,newKey=%v", head.ID(), oldKey, newKey)
				clientMutex.Lock() //R
				rlock++
				ch, ok := clientsMap[head.ID()]
				rlock--
				clientMutex.Unlock() /*R*/
				if ok {
					if data.ID() == ch.ID && ch.Key == oldKey {
						closeClient(data.ID())
					} else { //非法客户端连接，断开
						fmt.Printf("invalid user %v has connected, abort.", head.ID())
						break
					}
				}
				first = false
				//忽略第一条心跳消息
				continue
			}
			i := 0
			for ; i < WAIT_LAST_CONN_TIMEOUT; i++ {
				clientMutex.Lock() //R
				rlock++
				_, ok := clientsMap[head.ID()]
				rlock--
				clientMutex.Unlock() /*R*/
				if ok {
					fmt.Printf("waiting last connection closed of user %v, %v seconds.", head.ID(), i)
					time.Sleep(1 * time.Second)
				} else {
					break
				}
			}
			if i == WAIT_LAST_CONN_TIMEOUT {
				fmt.Printf("waiting last connection closed of user %v timeout.", head.ID())
				break
			}
			uid = head.ID()
			ch := service.NewServiceConn(head.ID(), conn, CLIENT_CHANNEL_SIZE, newKey)
			clientMutex.Lock()
			wlock++
			clientsMap[head.ID()] = ch
			wlock--
			clientMutex.Unlock()
			conn.SetReadTimeout(clientReadTimeout)
			go WriteMsg(ch, conn)
			defer closeClient(head.ID())
			second = false
		}
		if head.ID() != uid { //非法消息
			fmt.Printf("invalid message, uid not match. expect %v, but is %v", uid, head.ID())
			break
		}
		if data.Type() == 65532 {
			SendHeart2(data.ID())
		} else if data.Type() != service.HEARTBEAT_MSG {
			servers, ok := servicesMap[data.Type()]
			if !ok {
				glog.Errorf("invalid message type %v", data.Type())
				break
			}

			id := data.ID() % uint32(len(servers))
			select {
			case servers[id].Ch <- data:
			default:
				glog.Warningf("Server Channel %v Full : %v\n", id, len(servers[id].Ch))
				dropServerMsgs++
			}
		}
	}
}

func SendHeart2(uid uint32) {
	data := service.NewMsg(0)
	data.SetID(uid)
	data.SetType(65532)
	clientMutex.Lock()
	ch, ok := clientsMap[uid]
	clientMutex.Unlock()
	if ok {
		//普通消息或断开连接消息
		select {
		case ch.Ch <- data:
		default:
			dropMsgs++
		}
	}
	return
}

func WriteMsg(ch *service.ServiceConn, conn *libnet.TCPConn) {
	defer conn.Close()
	atomic.AddInt32(&wt, 1)
	for {
		glog.Debugf("waiting to write to client %v ...", ch.ID)
		atomic.AddInt32(&waiting, 1)
		msg, ok := <-ch.Ch
		atomic.AddInt32(&waiting, -1)
		if !ok {
			glog.Warningf("User %v Channel Closed", ch.ID)
			break
		}

		if msg.Length() == 0 && msg.Type() != 65532 {
			glog.Debugf("User %v Connection Closed By Server", msg.ID())
			needSendCloseLock.Lock()
			needSendCloseMsg[msg.ID()] = false
			needSendCloseLock.Unlock()
			break
		}

		err := conn.WriteSafe([]byte(msg))
		if err != nil {
			glog.Errorf("user %v WriteSafe content error: %v", msg.ID(), err.Error())
			break
		}
		glog.Debugf("send to client ip=%v len=%d id=%d type=%d content=%s\n", ch.IP, msg.Length(), msg.ID(), msg.Type(), string(msg.Content()))
	}
	atomic.AddInt32(&wt, -1)
}

func ReadServerMsgs(conn *libnet.TCPConn, msgType uint16, id uint32) {
	defer conn.Close()
	conn.SetReadTimeout(SERVER_READ_TIMEOUT)

	head := service.NewMsg(0)
	for {
		err := conn.ReadSafe(head.Header())
		if err != nil {
			glog.Errorf("server %v-%v ReadSafe head error: %v\n", msgType, id, err.Error())
			break
		}

		data := service.NewMsg(head.Length())
		err = conn.ReadSafe(data.Content())
		if err != nil {
			glog.Errorf("server %v-%v ReadSafe content error: %v\n", msgType, id, err.Error())
			break
		}

		copy(data.Header(), head.Header())
		glog.Debugf("from server %v-%v len=%d id=%d type=%d content=%s\n", msgType, id, data.Length(), data.ID(), data.Type(), string(data.Content()))

		uid := head.ID()   // client id
		if uid > 0  {
			func() {
				clientMutex.Lock() //R
				rlock++
				defer func() { rlock--; clientMutex.Unlock() /*R*/ }()
				ch, ok := clientsMap[uid]
				if ok { //普通消息或断开连接消息
					select {
					case ch.Ch <- data:
					default: //不能阻塞协程，只能丢弃较早的消息
						dropMsgs++
					}
				} else { //该用户并未连接
					glog.Warningf("User %v Not Exists, Send To Default Chan.\n", uid)
					e := service.NewMsg(0)
					e.SetID(uid)
					e.SetType(0)
					e.SetLength(0)
					servicesMap[0][uid%uint32(len(servicesMap[0]))].Ch <- e // TODO use random
				}
			}()
		}
	}
}

func InitServerMsg() (msg service.Msg) {
	clientMutex.Lock() //R
	rlock++
	defer func() { rlock--; clientMutex.Unlock() /*R*/ }()
	msg = service.NewMsg(uint32(len(clientsMap) * 8))
	index := 0
	content := msg.Content()
	for _, ch := range clientsMap {
		copy(content[index:index+4], yh_utils.Uint32ToBytes(ch.ID))
		index += 4
		copy(content[index:index+4], ch.IP)
		index += 4
	}
	return
}

func WriteServerMsgs(msgType uint16, svc *service.ServiceConn) {
	ip := make([]byte, 4)
	var msg, msgBuffer service.Msg
	sent := true
	var conn *yh_net.TCPConn = nil
	atomic.AddInt32(&rts, 1)
	for {
		for {
			conn = svc.Conn()
			if sent == false {
				msg = msgBuffer
				glog.Debugf("server %v-%v use last msg %v", msgType, svc.ID, msg.Content())
			} else {
				msg = <-svc.Ch
				msgBuffer = msg
				sent = false
				clientMutex.Lock() //R
				rlock++
				ch, ok := clientsMap[msg.ID()]
				rlock--
				clientMutex.Unlock() /*R*/
				if ok {
					copy(ip, ch.IP)
				}
			}
			err := conn.WriteSafe(ip)
			if err != nil {
				glog.Errorf("server %v-%v Write IP error: %v", msgType, svc.ID, err.Error())
				time.Sleep(1 * time.Second)
				continue
			}

			err = conn.WriteSafe([]byte(msg))
			if err != nil {
				glog.Errorf("server %v-%v Write msg error: %v", msgType, svc.ID, err.Error())
				continue
			}
			sent = true
			glog.Debugf("send to server %v-%v ip=%v len=%d id=%d type=%d content=%s", msgType, svc.ID, ip, msg.Length(), msg.ID(), msg.Type(), string(msg.Content()))
			break
		}
	}
	atomic.AddInt32(&rts, -1)
}

// 建立server2server的连接
func handleServer(ip string, port int, msgType uint16, svc *service.ServiceConn) {
	fmt.Printf("connecting server %v-%v(%v:%v)....\n", msgType, svc.ID, ip, port)
	conn, e := libnet.Connect(ip, port)
	if e != nil {
		fmt.Printf("connect %v-%v(%v:%v) failed : %v\n", msgType, svc.ID, ip, port, e.Error())
		time.Sleep(5 * time.Second)
		return
	}

	fmt.Printf("server %v-%v(%v:%v) connected.\n", msgType, svc.ID, ip, port)
	//msg := InitServerMsg()
	//err = conn.WriteSafe(make([]byte, 4))
	//if err != nil {
	//	fmt.Printf("server %v-%v Write IP error: %v\n", msgType, aconn.ID, err.Error())
	//	continue
	//}
	//
	//err = conn.WriteSafe([]byte(msg))
	//if err != nil {
	//	fmt.Printf("server %v-%v Write msg error: %v\n", msgType, aconn.ID, err.Error())
	//	continue
	//}
	//b_buf := bytes.NewBuffer(msg.Content())
	//ids := make([]uint32, 0, 5)
	//for i := 0; i < len(msg.Content()); i += 4 {
	//	var id uint32
	//	e := binary.Read(b_buf, binary.BigEndian, &id)
	//	if e != nil {
	//		fmt.Printf("Read Error %v\n", e.Error())
	//		break
	//	}
	//	ids = append(ids, id)
	//}
	//fmt.Sprintf("send to server %v-%v ip=0.0.0.0 len=%d id=%d type=%d content=%v", msgType, aconn.ID, msg.Length(), msg.ID(), msg.Type(), ids)
	svc.SetConn(conn)

	go WriteServerMsgs(msgType, svc)
	ReadServerMsgs(conn, msgType, svc.ID)
}

func InitServers(svlst []config.Service) (err error) {
	servicesMap = make(map[uint16][]*service.ServiceConn)
	for _, sv := range svlst {
		svcs, ok := servicesMap[sv.Type]
		if !ok {
			svcs = make([]*service.ServiceConn, 0, 4)
			servicesMap[sv.Type] = svcs
		}

		svc := service.NewServiceConn(uint32(len(svcs)), nil, SERVER_CHANNEL_SIZE, 0)
		servicesMap[sv.Type] = append(svcs, svc)

		go handleServer(sv.Ip, sv.Port, sv.Type, svc)
	}
	return
}

func Accept(ip string, port int) {
	ln, err := yh_net.Listen(ip, port)
	if err != nil {
		fmt.Println(err.Error())
		return
	}

	for {
		conn, err := ln.Accept()
		if err != nil {
			fmt.Println(err.Error())
			return
		}

		glog.Debugf("new connection from %v", conn.RemoteAddr())
		go ReadMsgs(conn)
	}
}

func main() {
	if len(os.Args) < 2 {
		fmt.Printf("invalid args : %s [config]\n", os.Args[0])
		return
	}

	rand.Seed(time.Now().UnixNano())
	conf, e := yh_config.NewConfig(os.Args[1])
	if e != nil {
		fmt.Println(e.Error())
		return
	}

    var ps int = 2
	var lv int = 0

	runtime.GOMAXPROCS(ps)
	glog  = fmtlog.NewFmtLog(lv, "adapter", true)
	defer glog.Close()

	go logStatus()

	e = InitServers()
	if e != nil {
		glog.Errorf("init servers error : %v", e)
		return
	}

	port, err := strconv.Atoi(conf.Items["port"])
	if err != nil {
		glog.Errorf(err.Error())
		return
	}

	ips := strings.Split(conf.Items["ips"], ",")
	for _, ip := range ips {
		go Accept(ip, port)
	}

	defer func() {
		if err := recover(); err != nil {
			debug.PrintStack()
		}
	}()

	for {
		time.Sleep(30 * time.Second)
		glog.Debugf("now client num: %d", cur_conn_num)
	}
}
